package com.headfirst.dmp.report.tags

import ch.hsr.geohash.GeoHash
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.Row
import redis.clients.jedis.JedisCluster

/**
  *  打商圈标签
  */
object Tags4BusinessTag extends TagsTrait {
  /**
    * 打商圈标签方法
    *
    * @param args
    * @return
    */
  override def makeTags(args: Any*): Map[String, Int] = {

    var map=Map[String,Int]()
    val row: Row = args(0).asInstanceOf[Row]
    val cluster: JedisCluster = args(1).asInstanceOf[JedisCluster]

    val lat: String = row.getAs[String]("lat")
    val longs: String = row.getAs[String]("longs")

    if(StringUtils.isNotBlank(lat) && StringUtils.isNotBlank(longs)){

      val lat2: Double = lat.toDouble
      val longs2: Double = longs.toDouble

      if(lat2 >=3.86 && lat2 <=53.55 && longs2>=73.66 && longs2 <=135.05){
        val base3: String = GeoHash.withCharacterPrecision(lat2,longs2,8).toBase32
        cluster.get(base3).foreach(bs => map += ("BS"+bs -> 1))
      }

    }

    map
  }

}
